package org.qy.star_client;

import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import org.qy.star_common.enums.ExchangeType;
import org.qy.star_common.exception.MQException;
import org.qy.star_common.message.Consumer;
import org.qy.star_common.message.MessageProperties;
import org.qy.star_grpc.core_api.BrokerCoreApiGrpc;
import org.qy.star_grpc.core_api.BrokerCoreApiProto;
import org.qy.star_grpc.core_api.BrokerCoreApiProto.*;

import java.io.IOException;
import java.util.UUID;

/**
 * @author yinger
 * @description TODO
 * @date 2024/3/30 16:00
 **/
public class Channel {
    private StreamObserver<Subscribe> requestStreamChannel;
    private ManagedChannel channel; // 访问所有方法共用此通道（除了订阅消息）
    private ManagedChannel subscribeChannel; // 订阅消息专用通道
    private ManagedChannel heartBeatChannel; // 心跳检测专用通道
    private String username;
    private String password;
    private String vhName;
    private final String consumerTag;

    public Channel(String consumerTag) {
        this.consumerTag = consumerTag;
    }

    // 给出一些api，直接调用服务端 核心 api
    private boolean userDeclare(String username, String password) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            BrokerCoreApiProto.User user = BrokerCoreApiProto.User.newBuilder().setUsername(username).setPassword(password).build();
            BrokerCoreApiProto.UnifiedResponse response = blockingStub.userDeclare(user);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 创建用户失败 ... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean userDeclare() {
        return userDeclare(username, password);
    }

    private boolean userUpdate(String username, String password, String newPassword) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            BrokerCoreApiProto.User user = BrokerCoreApiProto.User.newBuilder().setUsername(username).setPassword(password).setNewPassword(newPassword).build();
            BrokerCoreApiProto.UnifiedResponse response = blockingStub.userUpdate(user);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 修改密码失败 ... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean userUpdate(String newPassword) {
        return userUpdate(username, password, newPassword);
    }

    private boolean userDelete(String username, String password) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            User user = User.newBuilder().setUsername(username).setPassword(password).build();
            UnifiedResponse response = blockingStub.userDelete(user);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 删除用户失败 ... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean userDelete() {
        return userDelete(username, password);
    }

    private boolean virtualHostDeclare(String username, String password, String virtualHostName) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            VirtualHost virtualHost = VirtualHost.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).build();
            UnifiedResponse response = blockingStub.virtualHostDeclare(virtualHost);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 创建虚拟主机失败 ... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean virtualHostDeclare() {
        return virtualHostDeclare(username, password, vhName);
    }

    private boolean virtualHostDelete(String username, String password, String virtualHostName) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            VirtualHost virtualHost = VirtualHost.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).build();
            UnifiedResponse response = blockingStub.virtualHostDelete(virtualHost);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 删除虚拟主机失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean virtualHostDelete() {
        return virtualHostDelete(username, password, vhName);
    }

    private boolean exchangeDeclare(String username, String password, String virtualHostName, String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            int type = 0;
            if (exchangeType == ExchangeType.DIRECT) {
                type = 0;
            } else if (exchangeType == ExchangeType.FANOUT) {
                type = 1;
            } else if (exchangeType == ExchangeType.TOPIC) {
                type = 2;
            }
            ExchangeDeclare exchangeDeclare = ExchangeDeclare.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setExchangeName(exchangeName).setExchangeType(type).setDurable(durable).setAutoDelete(autoDelete).build();
            UnifiedResponse response = blockingStub.exchangeDeclare(exchangeDeclare);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 创建交换机失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete) {
        return exchangeDeclare(username, password, vhName, exchangeName, exchangeType, durable, autoDelete);
    }

    private boolean exchangeDelete(String username, String password, String virtualHostName, String exchangeName) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            ExchangeDelete exchangeDelete = ExchangeDelete.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setExchangeName(exchangeName).build();
            UnifiedResponse response = blockingStub.exchangeDelete(exchangeDelete);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 交换机删除失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean exchangeDelete(String exchangeName) {
        return exchangeDelete(username, password, vhName, exchangeName);
    }

    private boolean queueDeclare(String username, String password, String virtualHostName, String queueName, boolean durable, boolean exclusive, boolean autoDelete) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            QueueDeclare queueDeclare = QueueDeclare.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setQueueName(queueName).setDurable(durable).setExclusive(exclusive).setAutoDelete(autoDelete).build();
            UnifiedResponse response = blockingStub.queueDeclare(queueDeclare);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 创建队列失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete) {
        return queueDeclare(username, password, vhName, queueName, durable, exclusive, autoDelete);
    }

    private boolean queueDelete(String username, String password, String virtualHostName, String queueName) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            QueueDelete queueDelete = QueueDelete.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setQueueName(queueName).build();
            UnifiedResponse response = blockingStub.queueDelete(queueDelete);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 删除队列失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean queueDelete(String queueName) {
        return queueDelete(username, password, vhName, queueName);
    }

    private boolean queueBind(String username, String password, String virtualHostName, String exchangeName, String queueName, String bindingKey) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            Bind bind = Bind.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setExchangeName(exchangeName).setQueueName(queueName).setBindingKey(bindingKey).build();
            UnifiedResponse response = blockingStub.bind(bind);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 绑定队列失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    public boolean queueBind(String exchangeName, String queueName, String bindingKey) {
        return queueBind(username, password, vhName, exchangeName, queueName, bindingKey);
    }

    private boolean queueUnbind(String username, String password, String virtualHostName, String exchangeName, String queueName) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            UnBind unbind = UnBind.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setExchangeName(exchangeName).setQueueName(queueName).build();
            UnifiedResponse response = blockingStub.unBind(unbind);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 解绑队列失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean queueUnbind(String exchangeName, String queueName) {
        return queueUnbind(username, password, vhName, exchangeName, queueName);
    }

    private boolean publish(String username, String password, String virtualHostName, String exchangeName, String routingKey, String messageId, int deliveryMode, byte[] payload) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            Publish publish = Publish.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setExchangeName(exchangeName).setRoutingKey(routingKey).setMessageId(messageId).setDeliveryMode(deliveryMode).setPayload(ByteString.copyFrom(payload)).build();
            UnifiedResponse response = blockingStub.publish(publish);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 发布消息失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean publish(String exchangeName, String routingKey, int deliveryMode, byte[] payload) {
        return publish(username, password, vhName, exchangeName, routingKey, "M" + UUID.randomUUID(), deliveryMode, payload);
    }

    private boolean ack(String username, String password, String virtualHostName, String queueName, String messageId) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiBlockingStub blockingStub = BrokerCoreApiGrpc.newBlockingStub(channel);
            Ack ack = Ack.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setQueueName(queueName).setMessageId(messageId).build();
            UnifiedResponse response = blockingStub.ack(ack);
            if (!response.getSuccess()) {
                System.out.println("[BrokerGrpcClient] 消息确认失败... " + response.getFailMsg());
                return false;
            } else {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean ack(String queueName, String messageId) {
        return ack(username, password, vhName, queueName, messageId);
    }


    private boolean subscribe(String username, String password, String virtualHostName, String queueName, boolean autoAck, Consumer consumer) {
        if (channel == null) {
            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
        }
        try {
            BrokerCoreApiGrpc.BrokerCoreApiStub stub = BrokerCoreApiGrpc.newStub(subscribeChannel);
             requestStreamChannel =  stub.subscribe(new StreamObserver<Message>() {
                @Override
                public void onNext(Message message) {
                    try {
                        System.out.println("收到服务器发送来的消息 " + message.getPayload());
                        consumer.handleDelivery("", new MessageProperties(message.getMessageId(), message.getRoutingKey(), message.getDeliveryMode()), message.getPayload().toByteArray());
                    } catch (MQException | IOException e) {
                        throw new RuntimeException(e);
                    }
                }

                @Override
                public void onError(Throwable t) {
                    System.out.println("服务器状态异常 ... " + t.getMessage());
                }

                @Override
                public void onCompleted() {
                    System.out.println("服务器状态异常 ... ");
                }
            });
             Subscribe subscribe = Subscribe.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setQueueName(queueName).setConsumerTag(consumerTag).setAutoAck(autoAck).build();
             // 订阅
            requestStreamChannel.onNext(subscribe);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }


//    // 订阅队列 使用服务器流式 rpc 订阅后不断接收服务器推送的消息
//    private boolean subscribe(String username, String password, String virtualHostName, String queueName, boolean autoAck, Consumer consumer) {
//        if (subscribeChannel == null) {
//            System.out.println("[BrokerGrpcClient] Channel 未创建 调用失败 ... ");
//        }
//        try {
//            BrokerCoreApiGrpc.BrokerCoreApiStub stub = BrokerCoreApiGrpc.newStub(subscribeChannel);
//            Subscribe subscribe = Subscribe.newBuilder().setUsername(username).setPassword(password).setVirtualHostName(virtualHostName).setQueueName(queueName).setConsumerTag(consumerTag).setAutoAck(autoAck).build();
//            stub.subscribe(subscribe, new StreamObserver<Message>() {
//                @Override
//                public void onNext(Message message) {
//                    // 处理消息
//                    try {
//                        System.out.println("收到服务器发送来的消息 " + message.getPayload());
//                        consumer.handleDelivery("", new MessageProperties(message.getMessageId(), message.getRoutingKey(), message.getDeliveryMode()), message.getPayload().toByteArray());
//                    } catch (MQException e) {
//                        throw new RuntimeException(e);
//                    } catch (IOException e) {
//                        throw new RuntimeException(e);
//                    }
//                }
//
//                @Override
//                public void onError(Throwable t) {
//                    // 错误处理
//                    System.out.println("[BrokerGrpcClient] 订阅队列失败... " + t.getMessage());
//                    t.printStackTrace();
//                }
//
//                @Override
//                public void onCompleted() {
//                    System.out.println("[BrokerGrpcClient] 订阅队列完成... ");
//                }
//            });
//            // 客户端等待返回
//            subscribeChannel.awaitTermination(10, TimeUnit.DAYS);
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//        return false;
//    }

    public boolean subscribe(String queueName, boolean autoAck, Consumer consumer) {
        return subscribe(username, password, vhName, queueName, autoAck, consumer);
    }

    public static Channel channelFactory(String ip, int port) {
        Channel channel = new Channel(generateConsumerTar());
        channel.channel = ManagedChannelBuilder.forAddress(ip, port).usePlaintext().build();
        channel.heartBeatChannel = ManagedChannelBuilder.forAddress(ip, port).usePlaintext().build();
        channel.subscribeChannel = ManagedChannelBuilder.forAddress(ip, port).usePlaintext().build();
        return channel;
    }


    // 配置操作指向的用户和虚拟机
    public void configMQContext(String username, String password, String vhName) {
        this.username = username;
        this.password = password;
        this.vhName = vhName;
    }

    private static String generateConsumerTar() {
        return "C-" + UUID.randomUUID();
    }

//    public static void main(String[] args) {
//        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 7994).usePlaintext().build();
//
//        Channel c = new Channel("C-1");
//        c.channel = channel;
//        c.createConnection(c.consumerTag);
//
//        while (true) {
//            try {
//                Thread.sleep(1000);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
//        }
//    }
}